package cn.zwx.business.order.test.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @description: 顺序消费
 * @projectName:rocket-mq-example
 * @see:cn.zwx.business.order.test.order
 * @author:zhangwenxue
 * @createTime:2020/10/16 10:31
 * @version:1.0
 */
public class OrderMeaggessApp {

    public static void main(String[] args) throws InterruptedException {
        send();
        consumer();
    }



    /**
     * @description 顺序发送消息
     * 以购物流程为准 "创建订单">>"支付">>"发货">>"收货">>"五星好评"
     * 一个topic下有多个queue,为了保证发送有序,rocketmq提供了MessageQueueSelector队列选择机制,他有三种实现
     * SelectMessageQueueByHash 根据附加参数的Hash值，按照消息队列列表的大小取余数，得到消息队列的index
     * SelectMessageQueueByMachineRoom 开源的版本没有具体的实现，基本的目的应该是机器的就近原则分配
     * SelectMessageQueueByRandom 使用了简单的随机数选择算法
     * 我们可使用hash取模法,让同一个订单发送到同一个queue中,再使用同步发送,只有消息A发送成功,再发送消息B,这样,我们保证了发送有序.
     * rocketmq的topic内的队列机制,可以保证存储满足FIFO,剩下的只需要消费者顺序消费即可
     * @return void
     * @author zhangwenxue
     * @createTime 2020/10/16 10:33
     **/
    private static void send() {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("producerGroup1");
            producer.setNamesrvAddr("localhost:9876");
            producer.setRetryTimesWhenSendFailed(3);
            producer.start();
            String[] tags = new String[]{"创建订单", "支付", "发货", "收货", "五星好评"};
            for (int i = 5; i < 25; i++) {
                int orderId = i / 5;
                Message msg = new Message("OrderTopic1", tags[i % tags.length], "uniqueId:" + i,
                        ("order_" + orderId + " " + tags[i % tags.length]).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        //此刻arg == orderId,可以保证是每个订单进入同一个队列
                        Integer id = (Integer) arg;
//                        System.out.println("orderId:"+id);
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);
                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @description 顺序消费
     * 可以保证比如说 order4 是按照 "创建订单">>"支付">>"发货">>"收货">>"五星好评" 顺序的
     * 这里保证的是局部有序性
     * @return void
     * @author zhangwenxue
     * @createTime 2020/10/16 10:40
     **/
    private static void consumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_Consumer");
        consumer.setNamesrvAddr("localhost:9876");
        try {
            //设置Consumer从哪开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe("OrderTopic1", "*");
            // 实现了MessageListenerOrderly表示一个队列只会被一个线程取到, 第二个线程无法访问这个队列,MessageListenerOrderly默认单线程
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    try {
                        System.out.println("orderInfo: " + new String(msgs.get(0).getBody(), "utf-8"));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("Consumer1 Started.");
        while (true) {
        }
    }



}
